package com.platform.apex.util;

import java.io.Serializable;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

// elasticsearch exchange data bus , kafka consumer to inqeue, elasticsearch job to outqeue
public class KafkaMessageBus implements Serializable{

    private  static BlockingQueue<FlieNameAndContent> MESSAGE_BUS;

    
//    public KafkaMessageBus(int busSize) {
//		MESSAGE_BUS = new LinkedBlockingQueue<FlieNameAndContent>(busSize);
//	}
    
    
    public BlockingQueue<FlieNameAndContent> getQueue(){
    	return this.MESSAGE_BUS;
    }

	public static  void init(int busSize) {
        MESSAGE_BUS = new LinkedBlockingQueue<FlieNameAndContent>(busSize);
    }

    public static  void put(FlieNameAndContent info) throws InterruptedException {
        MESSAGE_BUS.put(info);
    }

    public static  FlieNameAndContent take() throws InterruptedException {
    	FlieNameAndContent msg = null;
        if (MESSAGE_BUS != null) {
            msg = MESSAGE_BUS.take();
        }
        return msg;
    }

    public  static FlieNameAndContent poll(long timeout, TimeUnit unit) throws InterruptedException {
    	FlieNameAndContent msg = null;
        if (MESSAGE_BUS != null) {
            msg = MESSAGE_BUS.poll(timeout, unit);
        }
        return msg;
    }

    public  static int size() {
        if (MESSAGE_BUS != null) {
            return MESSAGE_BUS.size();
        } else {
            return 0;
        }
    }
}
